package com.niit.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object CreatRDD {

  def main(args: Array[String]): Unit = {
    //1.准备环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark")
    //设置分区数                     设置分区数量4
    sparkConf.set("spark.default.parallelism","4")
    //sparkConf.set("spark.testing.memory","471859200")
    //2.让SparkContext去加载配置
    val sparkContext = new SparkContext(sparkConf)
    //3.从内存中创建RDD
    //memoryCreateRDD(sparkContext)

    //4.从外部(文件)创建RDD
    fileCreateRDD(sparkContext)

  }

  def memoryCreateRDD(sparkContext: SparkContext): Unit ={
    //创建RDD--从内存中创建RDD，将内存中集合的数据作为处理的数据源
    val rdd1:RDD[Int] = sparkContext.parallelize(1 to 10)//如果只写一个参数 那么就采用的是默认分区
    val rdd2:RDD[Int] = sparkContext.parallelize(1 to 10,3)//3: 分数区
    println(rdd1.getNumPartitions)//获得底层分区数
    println(rdd1.partitions.length)//获得底层分区数
    //将计算的结果保存在文件当中
    //rdd1.saveAsTextFile("output")

    //makeRDD 和 parallelize 是一样  parallelize是makeRDD的底层写法
    val rdd3:RDD[Int] = sparkContext.makeRDD(1 to 10)//将Scala的中的数组或集合 转换成一个RDD对象
    //rdd的迭代
    rdd3.collect().foreach(println)

    //RDD就可以理解为 mapReduce

    //关闭环境
    sparkContext.stop()

  }

  def fileCreateRDD(sparkContext: SparkContext):Unit={
    /*
    1.在项目目录上创建一个input文件夹，并在该文件夹中创建1.txt 在里面写一些东西
     */
    val rdd4:RDD[String] = sparkContext.textFile("input/1.txt")//已经进行了自动分片
    rdd4.collect().foreach(println)

    //关闭资源
    sparkContext.stop();
  }

}
